# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: db\data\access.py
__author__ = "sanyi"
from db.views.scans_extended import ScanExtendedViewRow, ScansExtendedViewTable
from db.views.targets_extended import TargetExtendedViewRow, TargetsExtendedViewTable
from sqlalchemy import and_ as sqlalchemy_and
from sqlalchemy import select as sql_select
from db.tables.groups import GroupRow, GroupsTable
from db.tables.locations import LocationRow, LocationsTable
from db.tables.reports import ReportsTable, ReportRow
from db.tables.scan_session_jobs import ScanSessionJobsTable, ScanSessionJobRow
from db.tables.scan_session_vulns import ScanSessionVulnRow, ScanSessionVulnsTable
from db.tables.scan_sessions import ScanSessionRow, ScanSessionsTable
from db.tables.scans import ScanRow, ScansTable
from db.tables.target2group import Target2GroupRow, Target2GroupTable
from db.tables.target_vulns import TargetVulnRow, TargetVulnsTable
from db.tables.targets import TargetRow, TargetsTable
from db.views.targets_excluded_hours import (
    TargetsWithExcludedHoursViewRow,
    TargetsWithExcludedHoursViewTable,
)
from db.tables.user2group import User2GroupRow, User2GroupTable
from db.views.locations_extended import (
    LocationExtendedViewRow,
    LocationsExtendedViewTable,
)


def apply_filters(query, filters):
    return query.where(sqlalchemy_and(*filters))


def get_distinct_targets_id_by_group(user):
    select_from = Target2GroupTable.join(
        User2GroupTable, User2GroupRow.group_id == Target2GroupRow.group_id
    )
    return (
        sql_select((Target2GroupRow.target_id,))
        .select_from(select_from)
        .where(User2GroupRow.user_id == user["user_id"])
    )


def generate_report_access_query(user, include_deleted=False):
    """
    :param user:
    :param include_deleted:
    :return:
    """
    select_from = ReportsTable
    tail_filters = []
    if not include_deleted:
        tail_filters.append(ReportRow.deleted_at.is_(None))
    if user["owner_id"] is None:
        tail_filters.append(ReportRow.owner_id == user["user_id"])
    else:
        if user["access_all_groups"]:
            tail_filters.append(ReportRow.owner_id == user["owner_id"])
        else:
            tail_filters.append(ReportRow.creator_id == user["user_id"])
        return (select_from, tail_filters)


def generate_scan_session_query(user, joined_tables=None, include_deleted=False):
    """
    Soft delete is checked directly on ScanSessionRow.deleted_at
    :param user:
    :param joined_tables:
    :param include_deleted:
    :return:
    """
    select_from = ScanSessionsTable.join(
        ScansTable, ScanRow.scan_id == ScanSessionRow.scan_id
    )
    if joined_tables is not None:
        joined_tables.append(ScanSessionsTable)
    tail_filters = []
    if not include_deleted:
        tail_filters.append(ScanSessionRow.deleted_at.is_(None))
    if user["owner_id"] is None:
        tail_filters.append(ScanRow.owner_id == user["user_id"])
    else:
        if user["access_all_groups"]:
            tail_filters.append(ScanRow.owner_id == user["owner_id"])
        else:
            tail_filters.append(
                ScanRow.target_id.in_(get_distinct_targets_id_by_group(user))
            )
        return (select_from, tail_filters)


def generate_scan_session_job_query(user, joined_tables=None, include_deleted=False):
    """
    Soft delete is checked directly on ScanSessionRow.deleted_at
    :param user:
    :param joined_tables:
    :param include_deleted:
    :return:
    """
    select_from = ScanSessionJobsTable.join(
        ScanSessionsTable,
        ScanSessionJobRow.scan_session_id == ScanSessionRow.scan_session_id,
    ).join(ScansTable, ScanRow.scan_id == ScanSessionRow.scan_id)
    if joined_tables is not None:
        joined_tables.append(ScanSessionsTable, ScansTable)
    tail_filters = []
    if not include_deleted:
        tail_filters.append(ScanSessionRow.deleted_at.is_(None))
    if user["owner_id"] is None:
        tail_filters.append(ScanRow.owner_id == user["user_id"])
    else:
        if user["access_all_groups"]:
            tail_filters.append(ScanRow.owner_id == user["owner_id"])
        else:
            tail_filters.append(
                ScanRow.target_id.in_(get_distinct_targets_id_by_group(user))
            )
        return (select_from, tail_filters)


def generate_scan_query(user, include_deleted=False):
    """
    Soft delete is checked directly on ScanRow.deleted_at
    :param user:
    :param include_deleted:
    :return:
    """
    tail_filters = []
    if not include_deleted:
        tail_filters.append(ScanRow.deleted_at.is_(None))
    if user["owner_id"] is None:
        tail_filters.append(ScanRow.owner_id == user["user_id"])
    else:
        if user["access_all_groups"]:
            tail_filters.append(ScanRow.owner_id == user["owner_id"])
        else:
            tail_filters.append(
                ScanRow.target_id.in_(get_distinct_targets_id_by_group(user))
            )
        return (ScansTable, tail_filters)


def generate_scan_session_locations_query(
    user, joined_tables=None, include_deleted=False
):
    """
    Soft delete is checked ScanSessionRow.deleted_at
    :param user:
    :param joined_tables:
    :param include_deleted:
    :return:
    """
    select_from = LocationsTable.join(
        ScanSessionsTable, LocationRow.scan_session_id == ScanSessionRow.scan_session_id
    )
    select_from = select_from.join(
        TargetsTable, TargetRow.target_id == ScanSessionRow.target_id
    )
    if joined_tables is not None:
        joined_tables.append(TargetsTable)
        joined_tables.append(ScanSessionsTable)
    tail_filters = []
    if not include_deleted:
        tail_filters.append(ScanSessionRow.deleted_at.is_(None))
    if user["owner_id"] is None:
        tail_filters.append(TargetRow.owner_id == user["user_id"])
    else:
        if user["access_all_groups"]:
            tail_filters.append(TargetRow.owner_id == user["owner_id"])
        else:
            tail_filters.append(
                TargetRow.target_id.in_(get_distinct_targets_id_by_group(user))
            )
        return (select_from, tail_filters)


def generate_scan_session_locations_extended_query(
    user, joined_tables=None, include_deleted=False
):
    """
    Soft delete is checked ScanSessionRow.deleted_at
    :param user:
    :param joined_tables:
    :param include_deleted:
    :return:
    """
    select_from = LocationsExtendedViewTable.join(
        ScanSessionsTable,
        LocationExtendedViewRow.scan_session_id == ScanSessionRow.scan_session_id,
    )
    select_from = select_from.join(
        TargetsTable, TargetRow.target_id == ScanSessionRow.target_id
    )
    if joined_tables is not None:
        joined_tables.append(TargetsTable)
        joined_tables.append(ScanSessionsTable)
    tail_filters = []
    if not include_deleted:
        tail_filters.append(ScanSessionRow.deleted_at.is_(None))
    if user["owner_id"] is None:
        tail_filters.append(TargetRow.owner_id == user["user_id"])
    else:
        if user["access_all_groups"]:
            tail_filters.append(TargetRow.owner_id == user["owner_id"])
        else:
            tail_filters.append(
                TargetRow.target_id.in_(get_distinct_targets_id_by_group(user))
            )
        return (select_from, tail_filters)


def generate_scan_session_vulns_query(user, joined_tables=None, include_deleted=False):
    """
    Soft delete is checked ScanSessionRow.deleted_at
    :param user:
    :param joined_tables:
    :param include_deleted:
    :return:
    """
    select_from = ScanSessionVulnsTable
    select_from = select_from.join(
        ScanSessionsTable,
        ScanSessionRow.scan_session_id == ScanSessionVulnRow.scan_session_id,
    )
    select_from = select_from.join(
        TargetsTable, TargetRow.target_id == ScanSessionRow.target_id
    )
    if joined_tables is not None:
        joined_tables.append(TargetsTable)
        joined_tables.append(ScanSessionsTable)
    tail_filters = []
    if not include_deleted:
        tail_filters.append(ScanSessionRow.deleted_at.is_(None))
    if user["owner_id"] is None:
        tail_filters.append(TargetRow.owner_id == user["user_id"])
    else:
        if user["access_all_groups"]:
            tail_filters.append(TargetRow.owner_id == user["owner_id"])
        else:
            tail_filters.append(
                TargetRow.target_id.in_(get_distinct_targets_id_by_group(user))
            )
        return (select_from, tail_filters)


def generate_target_vulns_query(user, joined_tables=None, include_deleted=False):
    """
    Soft delete is checked on TargetRow.deleted_at
    :param user:
    :param joined_tables:
    :param include_deleted:
    :return:
    """
    select_from = TargetVulnsTable.join(
        TargetsTable, TargetRow.target_id == TargetVulnRow.target_id
    )
    if joined_tables is not None:
        joined_tables.append(TargetsTable)
    tail_filters = []
    if not include_deleted:
        tail_filters.append(TargetRow.deleted_at.is_(None))
    if user["owner_id"] is None:
        tail_filters.append(TargetRow.owner_id == user["user_id"])
    else:
        if user["access_all_groups"]:
            tail_filters.append(TargetRow.owner_id == user["owner_id"])
        else:
            tail_filters.append(
                TargetRow.target_id.in_(get_distinct_targets_id_by_group(user))
            )
        return (select_from, tail_filters)


def generate_scan_ext_query(user, include_deleted=False):
    """
    Soft delete is checked on ScanExtendedViewRow.deleted_at
    :param user:
    :param include_deleted:
    :return:
    """
    tail_filters = []
    if not include_deleted:
        tail_filters.append(ScanExtendedViewRow.deleted_at.is_(None))
    if user["owner_id"] is None:
        tail_filters.append(ScanExtendedViewRow.owner_id == user["user_id"])
    else:
        if user["access_all_groups"]:
            tail_filters.append(ScanExtendedViewRow.owner_id == user["owner_id"])
        else:
            tail_filters.append(
                ScanExtendedViewRow.target_id.in_(
                    get_distinct_targets_id_by_group(user)
                )
            )
        return (ScansExtendedViewTable, tail_filters)


def generate_target_xh_query(user, include_deleted=False):
    """
    Soft delete is checked on TargetRow.deleted_at
    :param user:
    :param include_deleted:
    :return:
    """
    tail_filters = []
    if not include_deleted:
        tail_filters.append(TargetsWithExcludedHoursViewRow.deleted_at.is_(None))
    if user["owner_id"] is None:
        tail_filters.append(TargetsWithExcludedHoursViewRow.owner_id == user["user_id"])
    else:
        if user["access_all_groups"]:
            tail_filters.append(
                TargetsWithExcludedHoursViewRow.owner_id == user["owner_id"]
            )
        else:
            tail_filters.append(
                TargetsWithExcludedHoursViewRow.target_id.in_(
                    get_distinct_targets_id_by_group(user)
                )
            )
        return (TargetsWithExcludedHoursViewTable, tail_filters)


def generate_target_query(user, include_deleted=False):
    """
    Soft delete is checked on TargetRow.deleted_at
    :param user:
    :param include_deleted:
    :return:
    """
    tail_filters = []
    if not include_deleted:
        tail_filters.append(TargetRow.deleted_at.is_(None))
    if user["owner_id"] is None:
        tail_filters.append(TargetRow.owner_id == user["user_id"])
    else:
        if user["access_all_groups"]:
            tail_filters.append(TargetRow.owner_id == user["owner_id"])
        else:
            tail_filters.append(
                TargetRow.target_id.in_(get_distinct_targets_id_by_group(user))
            )
        return (TargetsTable, tail_filters)


def generate_target_ext_query(user, include_deleted=False):
    """
    Soft delete is checked on TargetExtendedViewRow.deleted_at
    :param user:
    :param include_deleted:
    :return:
    """
    tail_filters = []
    if not include_deleted:
        tail_filters.append(TargetExtendedViewRow.deleted_at.is_(None))
    if user["owner_id"] is None:
        tail_filters.append(TargetExtendedViewRow.owner_id == user["user_id"])
    else:
        if user["access_all_groups"]:
            tail_filters.append(TargetExtendedViewRow.owner_id == user["owner_id"])
        else:
            tail_filters.append(
                TargetExtendedViewRow.target_id.in_(
                    get_distinct_targets_id_by_group(user)
                )
            )
        return (TargetsExtendedViewTable, tail_filters)


def generate_group_query(user, joined_tables=None):
    """
    Groups don't have soft delete
    :param user:
    :param joined_tables:
    :return:
    """
    if user["owner_id"] is None:
        return (GroupsTable, [GroupRow.owner_id == user["user_id"]])
    elif user["access_all_groups"]:
        return (GroupsTable, [GroupRow.owner_id == user["owner_id"]])
    else:
        select_from = GroupsTable.join(
            User2GroupTable, User2GroupRow.group_id == GroupRow.group_id
        )
        if joined_tables is not None:
            joined_tables.append(User2GroupTable)
        return (select_from, [User2GroupRow.user_id == user["user_id"]])


def generate_target2group_query(user, joined_tables=None):
    """
    Groups don't have soft delete
    :param user:
    :param joined_tables:
    :return:
    """
    if user["owner_id"] is None:
        return (Target2GroupTable, [Target2GroupRow.owner_id == user["user_id"]])
    elif user["access_all_groups"]:
        return (Target2GroupTable, [Target2GroupRow.owner_id == user["owner_id"]])
    else:
        select_from = Target2GroupTable.join(
            User2GroupTable, User2GroupRow.group_id == Target2GroupRow.group_id
        )
        if joined_tables is not None:
            joined_tables.append(Target2GroupTable)
        return (select_from, [User2GroupRow.user_id == user["user_id"]])
